// AnomalyDetectionReducer.java
package com.taxi.analysis.anomalyDetection;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * 高价行程异常检测Reducer
 * 功能：聚合和统计各类异常行程
 * 技术：自定义分组聚合 + 统计分析
 */
public class AnomalyDetectionReducer extends Reducer<Text, Text, Text, Text> {

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        // 解析复合键
        String[] keyParts = key.toString().split("_");
        if (keyParts.length < 2) {
            return;
        }

        String anomalyType = keyParts[0];
        String regionGroup = keyParts[1];

        // 统计变量初始化
        int totalCount = 0;
        double totalFare = 0.0;
        double totalDistance = 0.0;
        double totalFarePerMile = 0.0;
        int totalPassengers = 0;

        // 价格分组统计
        Map<String, Integer> priceGroupCounts = new HashMap<>();

        // 地区详细统计
        Map<String, Integer> pickupZoneCounts = new HashMap<>();
        Map<String, Integer> dropoffZoneCounts = new HashMap<>();

        // 存储样本数据（用于详细分析）
        List<String> sampleData = new ArrayList<>();

        // 遍历所有异常记录进行聚合
        for (Text value : values) {
            try {
                String record = value.toString();
                String[] fields = record.split("\\|");

                // 解析记录字段
                // 格式: pickupZone|dropoffZone|tripDistance|totalAmount|passengerCount|farePerMile|priceGroup|anomalyType
                if (fields.length >= 8) {
                    String pickupZone = fields[0];
                    String dropoffZone = fields[1];
                    double tripDistance = parseDouble(fields[2]);
                    double totalAmount = parseDouble(fields[3]);
                    int passengerCount = parseInt(fields[4]);
                    double farePerMile = parseDouble(fields[5]);
                    String priceGroup = fields[6];

                    // 累计统计
                    totalCount++;
                    totalFare += totalAmount;
                    totalDistance += tripDistance;
                    totalFarePerMile += farePerMile;
                    totalPassengers += passengerCount;

                    // 价格分组统计
                    priceGroupCounts.put(priceGroup,
                            priceGroupCounts.getOrDefault(priceGroup, 0) + 1);

                    // 地区统计
                    pickupZoneCounts.put(pickupZone,
                            pickupZoneCounts.getOrDefault(pickupZone, 0) + 1);
                    dropoffZoneCounts.put(dropoffZone,
                            dropoffZoneCounts.getOrDefault(dropoffZone, 0) + 1);

                    // 保存前10个样本用于详细分析
                    if (sampleData.size() < 10) {
                        sampleData.add(record);
                    }
                }

            } catch (Exception e) {
                System.err.println("处理记录时出错: " + value.toString());
            }
        }

        // 计算平均值和统计指标
        if (totalCount > 0) {
            double avgFare = totalFare / totalCount;
            double avgDistance = totalDistance / totalCount;
            double avgFarePerMile = totalFarePerMile / totalCount;
            double avgPassengers = (double) totalPassengers / totalCount;

            // 找出最频繁的上车和下车地点
            String topPickupZone = findTopZone(pickupZoneCounts);
            String topDropoffZone = findTopZone(dropoffZoneCounts);
            String topPriceGroup = findTopZone(priceGroupCounts);

            // 构建详细的统计报告
            StringBuilder report = new StringBuilder();

            // 基础统计信息
            report.append(String.format("异常类型:%s|", anomalyType));
            report.append(String.format("区域:%s|", regionGroup));
            report.append(String.format("总数量:%d|", totalCount));
            report.append(String.format("平均费用:%.2f|", avgFare));
            report.append(String.format("平均距离:%.2f|", avgDistance));
            report.append(String.format("平均单价:%.2f|", avgFarePerMile));
            report.append(String.format("平均乘客:%.1f|", avgPassengers));

            // 热点地区信息
            report.append(String.format("热点上车地:%s|", topPickupZone));
            report.append(String.format("热点下车地:%s|", topDropoffZone));
            report.append(String.format("主要价格组:%s|", topPriceGroup));

            // 严重程度评估
            String severity = assessSeverity(anomalyType, avgFare, avgFarePerMile, totalCount);
            report.append(String.format("严重程度:%s", severity));

            // 输出统计结果
            context.write(new Text(String.format("异常统计_%s_%s", anomalyType, regionGroup)),
                    new Text(report.toString()));

            // 输出详细样本（用于进一步分析）
            if (!sampleData.isEmpty()) {
                StringBuilder samples = new StringBuilder();
                samples.append("样本数据:");
                for (int i = 0; i < Math.min(5, sampleData.size()); i++) {
                    samples.append(sampleData.get(i)).append(";");
                }

                context.write(new Text(String.format("异常样本_%s_%s", anomalyType, regionGroup)),
                        new Text(samples.toString()));
            }
        }
    }

    /**
     * 找出出现频率最高的地区或分组
     */
    private String findTopZone(Map<String, Integer> zoneCounts) {
        String topZone = "未知";
        int maxCount = 0;

        for (Map.Entry<String, Integer> entry : zoneCounts.entrySet()) {
            if (entry.getValue() > maxCount) {
                maxCount = entry.getValue();
                topZone = entry.getKey();
            }
        }

        return String.format("%s(%d次)", topZone, maxCount);
    }

    /**
     * 评估异常的严重程度
     */
    private String assessSeverity(String anomalyType, double avgFare,
                                  double avgFarePerMile, int count) {

        int severityScore = 0;

        // 根据异常类型评分
        if (anomalyType.contains("极高价")) {
            severityScore += 3;
        } else if (anomalyType.contains("高价")) {
            severityScore += 2;
        } else if (anomalyType.contains("统计异常")) {
            severityScore += 1;
        }

        // 根据平均费用评分
        if (avgFare >= 200) {
            severityScore += 3;
        } else if (avgFare >= 100) {
            severityScore += 2;
        } else if (avgFare >= 50) {
            severityScore += 1;
        }

        // 根据单价评分
        if (avgFarePerMile >= 15) {
            severityScore += 2;
        } else if (avgFarePerMile >= 10) {
            severityScore += 1;
        }

        // 根据数量评分
        if (count >= 1000) {
            severityScore += 2;
        } else if (count >= 100) {
            severityScore += 1;
        }

        // 确定严重程度等级
        if (severityScore >= 8) {
            return "极高风险";
        } else if (severityScore >= 6) {
            return "高风险";
        } else if (severityScore >= 4) {
            return "中等风险";
        } else if (severityScore >= 2) {
            return "低风险";
        } else {
            return "正常";
        }
    }

    /**
     * 安全的double解析
     */
    private double parseDouble(String str) {
        try {
            return Double.parseDouble(str.trim());
        } catch (NumberFormatException e) {
            return 0.0;
        }
    }

    /**
     * 安全的int解析
     */
    private int parseInt(String str) {
        try {
            return Integer.parseInt(str.trim());
        } catch (NumberFormatException e) {
            return 0;
        }
    }
}
